package com.rem.flink.flink7Stream;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 对流的操作
 * 合流 ->连接流处理 CoProcessFunction
 * 协同处理函数
 * <p>
 * 连接流处理 两个流，类型可以不同 处理数据
 *
 * @author Rem
 * @date 2022-10-20
 */

public class CoProcessFunctionTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        /**
         * 实时对账处理
         * 两个端的数据来源对账  成功的数据才会匹配对账成功
         */
        SingleOutputStreamOperator<Tuple3<String, String, Long>> appStream = env.fromElements(
                Tuple3.of("order-1", "app", 1000L),
                Tuple3.of("order-2", "app", 2000L)
        ).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, String, Long>>forMonotonousTimestamps()
                .withTimestampAssigner((element, recordTimestamp) -> element.f2));


        SingleOutputStreamOperator<Tuple4<String, String, String, Long>> thirdpartStream = env.fromElements(
                Tuple4.of("order-1", "third", "success", 3000L),
                Tuple4.of("order-3", "third", "success", 4000L)
        ).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple4<String, String, String, Long>>forMonotonousTimestamps()
                .withTimestampAssigner((element, recordTimestamp) -> element.f3));

        appStream.connect(thirdpartStream)
                .keyBy(a -> a.f0, a -> a.f0)

                /**
                 * 处理两个流的元素并产生单个输出流的函数
                 * 检测流中数据是否匹配
                 */
                .process(new CoProcessFunction<Tuple3<String, String, Long>, Tuple4<String, String, String, Long>, String>() {
                    // 定义状态变量，用来保存已经到达的事件
                    private ValueState<Tuple3<String, String, Long>> appEventState;
                    private ValueState<Tuple4<String, String, String, Long>> thirdPartyEventState;

                    @Override
                    public void open(Configuration parameters) {
                        appEventState = getRuntimeContext().getState(new ValueStateDescriptor<>("app-event", Types.TUPLE(Types.STRING, Types.STRING, Types.LONG)));
                        thirdPartyEventState = getRuntimeContext().getState(new ValueStateDescriptor<>("thirdparty-event", Types.TUPLE(Types.STRING, Types.STRING, Types.STRING, Types.LONG)));
                    }

                    @Override
                    public void processElement1(Tuple3<String, String, Long> value, Context ctx, Collector<String> out) throws Exception {
                        //在第一条流中 校验第二条流中的数据是否来过
                        if (thirdPartyEventState.value() != null) {
                            out.collect("对账成功1：" + "appStream:" + value + "  thirdpartStream:" + thirdPartyEventState.value());
                            // 清空状态
                            thirdPartyEventState.clear();
                        } else {
                            //更新状态
                            appEventState.update(value);
                            //注册5s定时器，等待另一条流
                            ctx.timerService().registerEventTimeTimer(value.f2 + 5000L);
                        }
                    }

                    @Override
                    public void processElement2(Tuple4<String, String, String, Long> value, Context ctx, Collector<String> out) throws Exception {
                        //在第二条流中 校验第一条流中的数据是否来过
                        if (appEventState.value() != null) {
                            out.collect("对账成功2：" + "thirdpartStream:" + value + "   appStream:" + appEventState.value());
                            // 清空状态
                            appEventState.clear();
                        } else {
                            // 清空状态
                            thirdPartyEventState.update(value);
                            //注册5s定时器，等待另一条流
                            ctx.timerService().registerEventTimeTimer(value.f3 + 5000L);
                        }
                    }

                    /**
                     * 触发器
                     * @param timestamp
                     * @param ctx
                     * @param out
                     * @throws Exception
                     */
                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        // 定时器触发，判断状态，如果某个状态不为空，说明另一条流中事件没来
                        if (appEventState.value() != null) {
                            out.collect("对账失败1：" + appEventState.value() + "  " + "第三方支付平台信息未到");
                        }
                        if (thirdPartyEventState.value() != null) {
                            out.collect("对账失败2：" + thirdPartyEventState.value() + "  " + "app信息未到");
                        }
                        appEventState.clear();
                        thirdPartyEventState.clear();
                    }
                }).print();


        env.execute();
    }

}
